/*
 *
 *  * | Licensed 未经许可不能去掉「OPENIITA」相关版权
 *  * +----------------------------------------------------------------------
 *  * | Author: xw2sy@163.com
 *  * +----------------------------------------------------------------------
 *
 *  Copyright [2024] [OPENIITA]
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 * /
 */
package cc.iotkit.temporal.es.service;

import cc.iotkit.common.utils.MapstructUtils;
import cc.iotkit.data.manager.IDeviceInfoData;
import cc.iotkit.model.device.DeviceInfo;
import cc.iotkit.model.device.message.DeviceConsumePower;
import cc.iotkit.model.device.message.DeviceProperty;
import cc.iotkit.model.device.message.DevicePropertyCache;
import cc.iotkit.temporal.IDevicePropertyData;
import cc.iotkit.temporal.es.document.DocDeviceConsumePower;
import cc.iotkit.temporal.es.document.DocDeviceProperty;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.bucket.histogram.LongBounds;
import org.elasticsearch.search.aggregations.bucket.histogram.ParsedDateHistogram;
import org.elasticsearch.search.aggregations.metrics.ParsedSum;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.domain.Pageable;
import org.springframework.data.elasticsearch.core.ElasticsearchAggregations;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.document.Document;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;

import java.time.ZonedDateTime;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class DevicePropertyDataImpl implements IDevicePropertyData {

    @Autowired
    private ElasticsearchRestTemplate template;

    @Autowired
    @Qualifier("deviceInfoDataCache")
    private IDeviceInfoData deviceInfoData;

    private final Set<String> indexSet = new HashSet<>();

    @Override
    public List<DeviceProperty> findDevicePropertyHistory(String deviceId, String name, long start, long end, int size) {
        String index = getIndex(deviceId, name);
        NativeSearchQuery query = new NativeSearchQueryBuilder()
                .withQuery(
                        QueryBuilders.boolQuery()
                                .must(QueryBuilders.termQuery("deviceId", deviceId))
                                .must(QueryBuilders.rangeQuery("time")
                                        .from(start, true).to(end, true))

                )
                .withSorts(new FieldSortBuilder("time").order(SortOrder.DESC))
                .withPageable(Pageable.ofSize(size).withPage(0))
                .build();
        SearchHits<DocDeviceProperty> result = template.search(query, DocDeviceProperty.class, IndexCoordinates.of(index));
        return result.getSearchHits().stream()
                .map(h -> MapstructUtils.convert(h.getContent(), DeviceProperty.class))
                .collect(Collectors.toList());
    }

    /**
     * 统计指定用户(uid)在时间范围内的总能耗（不分桶）
     * @param uid
     * @param start
     * @param end
     * @return
     */
    public DeviceConsumePower getTotalEnergyConsumptionTrend(String uid, long start, long end) {
        NativeSearchQuery query = new NativeSearchQueryBuilder()
                .withQuery(
                        QueryBuilders.boolQuery()
                                .must(QueryBuilders.termQuery("uid", uid))
                                .must(QueryBuilders.rangeQuery("time")
                                        .from(start, true).to(end, false))
                )
                .withAggregations(AggregationBuilders.sum("totalPower").field("consumePower"))
                .build();

        SearchHits<DocDeviceConsumePower> hits = template.search(query, DocDeviceConsumePower.class);

        // 解析聚合结果
        Aggregations aggregations = ((ElasticsearchAggregations) hits.getAggregations()).aggregations();
        ParsedSum sum = aggregations.get("totalPower");

        return (sum != null) ? new DeviceConsumePower(sum.getValue()) : new DeviceConsumePower(0.0);
    }

    /**
     * 能耗统计聚合求和 用active_energy分组,按小时(calendarInterval=1h), 天(calendarInterval=1d), 月(calendarInterval=1m)
     */
    public List<DeviceConsumePower> getEnergyConsumptionTrend(String uid, String calendarInterval, long start, long end) {
        NativeSearchQuery query = new NativeSearchQueryBuilder()
                .withQuery(
                        QueryBuilders.boolQuery()
                                .must(QueryBuilders.termQuery("uid", uid))
                                .must(QueryBuilders.rangeQuery("time")
                                        .from(start, true).to(end, false))
                )
                .withAggregations(AggregationBuilders.dateHistogram("agg")
                        .field("time")
                        .calendarInterval(getDateInterval(calendarInterval))
                        .minDocCount(0)  // 补齐没有数据的桶
                        .extendedBounds(new LongBounds(start, end)) // 保证全时段覆盖
                        .subAggregation(AggregationBuilders.sum("sumPower").field("consumePower"))
                )
                .build();


        SearchHits<DocDeviceConsumePower> hits = template.search(query, DocDeviceConsumePower.class);

        // 解析聚合结果
        Aggregations aggregations = ((ElasticsearchAggregations) hits.getAggregations()).aggregations();
        ParsedDateHistogram histogram = aggregations.get("agg");

        List<DeviceConsumePower> result = new ArrayList<>();
        //Map<Long, Double> result = new TreeMap<>();

        if (histogram != null) {
            for (Histogram.Bucket bucket : histogram.getBuckets()) {
                if (Long.parseLong(bucket.getKeyAsString()) >= end) {
                    continue; // 按小时聚合时, 丢掉最后一个补出来的空桶
                }
                ParsedSum sum = bucket.getAggregations().get("sumPower");
                double value = (sum != null) ? sum.getValue() : 0.0;
                long time = ((ZonedDateTime) bucket.getKey()).toInstant().toEpochMilli();
                //result.put(time, value);
                result.add(new DeviceConsumePower(time, value));
            }
        }

        // ✅ 补齐缺失时间点
        /*long intervalMillis = getIntervalMillis(calendarInterval);
        for (long t = start; t <= end; t += intervalMillis) {
            result.putIfAbsent(t, 0.0);
        }*/

        return result;
    }

    private DateHistogramInterval getDateInterval(String calendarInterval) {
        //按类型聚合
        if ("h".equalsIgnoreCase(calendarInterval)) {
            return DateHistogramInterval.HOUR;
        } else if ("d".equalsIgnoreCase(calendarInterval)) {
            return DateHistogramInterval.DAY;
        } else if ("m".equalsIgnoreCase(calendarInterval)) {
            return DateHistogramInterval.MONTH;
        } else {
            // return DateHistogramInterval.hours(1);
            return DateHistogramInterval.HOUR;
        }
    }

    /**
     * 把 "1h" "1d" "1m" 转换为 ES 的 DateHistogramInterval
     */
    /*private DateHistogramInterval getDateInterval(String calendarInterval) {
        switch (calendarInterval) {
            case "1h": return DateHistogramInterval.HOUR;
            case "1d": return DateHistogramInterval.DAY;
            case "1m": return DateHistogramInterval.MONTH;
            default: throw new IllegalArgumentException("Unsupported interval: " + calendarInterval);
        }
    }*/

    /**
     * 获取毫秒级时间间隔
     */
    /*private long getIntervalMillis(String calendarInterval) {
        switch (calendarInterval) {
            case "1h": return Duration.ofHours(1).toMillis();
            case "1d": return Duration.ofDays(1).toMillis();
            case "1m": return Duration.ofDays(30).toMillis(); // 近似，按30天算
            default: throw new IllegalArgumentException("Unsupported interval: " + calendarInterval);
        }
    }*/

    /**
     * 计算最小时间间隔
     */
    private long getMinIntervalMillis(String calendarInterval) {
        switch (calendarInterval.toLowerCase()) {
            case "h": return 3600000L; // 1小时
            case "d": return 86400000L; // 1天
            case "m": return 2592000000L; // 30天近似
            default: return 3600000L; // 默认1小时
        }
    }

    @Override
    public void addProperties(String deviceId, Map<String, DevicePropertyCache> properties, long time) {
        properties.forEach((key, val) -> {
            DevicePropertyCache propertyCache = (DevicePropertyCache) val;
            String index = getIndex(deviceId, key);
            long occurred = Objects.nonNull(propertyCache.getOccurred()) ? propertyCache.getOccurred() : time;
            // 增加属性时加入判断, 如果属性是active_energy且有值, 则加入设备能耗统计, 需减去上次的值
            try {
                if ("active_energy".equals(key) && propertyCache.getValue() != null) {
                    addEnergy(deviceId, key, propertyCache, occurred);
                }
            } catch (Exception e) {
                log.error("deviceId: [{}]+propertyCache:{}, addProperties error: {} ,", deviceId, propertyCache, e.getMessage(), e);
            }
            template.save(
                    new DocDeviceProperty(UUID.randomUUID().toString(), deviceId, key, propertyCache.getValue(), occurred),
                    IndexCoordinates.of(index)
            );
        });
    }

    /**
     * 增加属性时加入判断, 如果属性是active_energy且有值, 则加入设备能耗统计, 需减去上次的值
     */
    public void addEnergy(String deviceId, String key, DevicePropertyCache propertyCache, long occurred) {
        String index = getIndex(deviceId, key);
        DeviceInfo deviceInfo = deviceInfoData.findByDeviceId(deviceId);
        String pk = deviceInfo.getProductKey();
        String uid = deviceInfo.getUid();
        if (uid == null) {return;}

        // 查询过去最新的一个数据
        NativeSearchQuery query = new NativeSearchQueryBuilder()
                .withQuery(
                        QueryBuilders.boolQuery()
                                .must(QueryBuilders.termQuery("deviceId", deviceId))

                )
                .withSorts(new FieldSortBuilder("time").order(SortOrder.DESC))
                .withPageable(Pageable.ofSize(1).withPage(0))
                .build();
        SearchHits<DocDeviceProperty> result = template.search(query, DocDeviceProperty.class, IndexCoordinates.of(index));
        long totalHits = result.getTotalHits();
        List<DeviceProperty> deviceProperties = result.getSearchHits().stream()
                .map(h -> MapstructUtils.convert(h.getContent(), DeviceProperty.class))
                .collect(Collectors.toList());
        if (totalHits > 0) {
            DeviceProperty lastDeviceProperty = deviceProperties.get(0);
            if (lastDeviceProperty != null) {
                /*Double lastEnergy = (Double) lastDeviceProperty.getValue();
                Double activeEnergy = (Double) propertyCache.getValue();*/
                Double lastEnergy = convertToDouble(lastDeviceProperty.getValue());
                Double activeEnergy = convertToDouble(propertyCache.getValue());
                Double consumePower;
                if (activeEnergy >= lastEnergy) {
                    consumePower = activeEnergy - lastEnergy;
                } else if (lastEnergy - activeEnergy < 10) {
                    consumePower = 0.0;
                } else {
                    consumePower = activeEnergy;
                }
                if (Math.abs(consumePower) < 1e-9) {
                    return;
                }
                // String consumePowerIndex = getIndex(deviceId, "consumePower3");
                template.save(
                        new DocDeviceConsumePower(UUID.randomUUID().toString(), deviceId, pk, consumePower, occurred, uid)
                );
            }

        }

    }

    private String getIndex(String deviceId, String name) {
        DeviceInfo deviceInfo = deviceInfoData.findByDeviceId(deviceId);
        if (deviceInfo == null) {
            return null;
        }
        String pk = deviceInfo.getProductKey();
        String index = String.format("device_property_%s_%s", pk, name).toLowerCase();
        if (!indexSet.contains(index)) {
            IndexCoordinates indexCoordinates = IndexCoordinates.of(index);
            if (!template.indexOps(indexCoordinates).exists()) {
                // 根据索引实体，获取mapping字段
                Document mapping = template.indexOps(indexCoordinates).createMapping(DocDeviceProperty.class);
                template.indexOps(indexCoordinates).create();
                // 创建索引mapping
                template.indexOps(indexCoordinates).putMapping(mapping);
            }
            indexSet.add(index);
        }
        return index;
    }

    /**
     * 安全地将对象转换为Double类型
     */
    private Double convertToDouble(Object value) {
        if (value == null) {
            return 0.0;
        }
        if (value instanceof Double) {
            return (Double) value;
        } else if (value instanceof Number) {
            return ((Number) value).doubleValue();
        } else {
            /*try {*/
                return Double.parseDouble(value.toString());
            /*} catch (NumberFormatException e) {
                return 0.0;
            }*/
        }
    }

}
